%%%====================================================================
%%% File    : pipe_mapred.erl
%%% Author  :  <vjache>
%%% Description: This is a map-reduce pipe implementation.
%%%
%%% Created : 27.03.2009
%%%====================================================================

-module(pipe_mapred).

%%
%% Include files
%%
-include("../include/pipe.hrl").
-include("../include/pgt.hrl").
-include("../include/util.hrl").
%%
%% Exported Functions
%%
-export([
		 new/2,
		 rcv_init/1,
		 test/0]).

%%====================================================================
%% API Functions
%%====================================================================


new(SenderSignature,NextPipe) ->
	LogPostfix=erlang:phash2(SenderSignature),
	%1. router pipe
	%2. buncher pipe
	%3. sender pipe
	SenderPipe=pipe_sender:new(
				 fun(Node)->
					   node2prc(Node,NextPipe)
				 end, 
				 SenderSignature),
	%LoggerPipe1=pipe_logger:new("after_bunch"++integer_to_list(LogPostfix)++".txt",SenderPipe),
	BunchPipe=pipe_folder:new(
				#folder{
						capacity=4000,
						result_fun=fun(Dict)-> dict:to_list(Dict)end,
						open_fun=fun()->dict:new()end,
						fold_fun=fun fold_fun/2
					   }, SenderPipe),
	%LoggerPipe2=pipe_logger:new("after_router"++integer_to_list(LogPostfix)++".txt",BunchPipe),
	svc_rrs:pipe_router(fun router_key_fun/1, BunchPipe).

router_key_fun(Record) 
  when is_tuple(Record) 
  andalso size(Record)>1
  andalso is_atom(element(1,Record))->
	element(2,Record);
router_key_fun(Record) ->
	exit({not_a_record,Record}).

fold_fun({N,V},Dict) ->
	dict:append(N, V, Dict);
fold_fun(BadArg,Dict) ->
	exit({badly_navigated,BadArg}).


rcv_init(Pipe) ->
	PipeState=pipe:open(Pipe),
	rcv_loop(PipeState).

%%--------------------------------------------------------------------
%% Local Functions
%%--------------------------------------------------------------------
node2prc(Node,NextPipe) ->
	pgt:ppg_obtain_worker(
	  #worker{
			  id={'$RCV-WRK-ID',erlang:phash2({Node,NextPipe})},%% TODO Fix to  {'$RCV-WRK-ID',Node,NextPipe#pipe.id}
			  spawn_spec={Node,?MODULE, rcv_init, [NextPipe]}
			 }).
rcv_loop(PipeStateIn) ->
	pgt:worker_receive_fold(
	  %% Message handler fun
	  fun
		 (?PIPE_SENDER_MSG(_SenderSignature,SenderPid,?EOF),PipeState) ->
			  SenderPid!?PIPE_SENDER_MSG_CONFIRM,
			  ?LOG_TRACE("MSG FUN",[]),
			  PipeState;
		 (?PIPE_SENDER_MSG(SenderSignature,SenderPid,Item),PipeState) when is_list(Item)->
			  SenderPid!?PIPE_SENDER_MSG_CONFIRM,
			  ?LOG_TRACE("PIPE_SENDER_MSG RECEIVED",[]),
			  lists:foldl(fun(I,PS)->
								  pipe:push(PS,{SenderSignature,I})
						  end, PipeState, Item); 
		 (UnexpectedMsg,PipeState) ->
			  ?LOG_TRACE("UNEXPECTED MSG ~p", [UnexpectedMsg]),
			  PipeState
	  end, 
	  %% Close fun
	  fun(PipeState)->
			  ?LOG_TRACE("CLOSE DIRECTIVE RECEIVED",[]),
			  pipe:close(PipeState),
			  ?LOG_TRACE("PIPE CLOSED",[])
	  end, PipeStateIn).




%%====================================================================
%% Test Functions
%%====================================================================
 
test() ->
	KeyFun=fun({K,V})->
				   K
		   end,
	WorkerId1=make_ref(),
	WorkerId2=make_ref(),
	ConsolePipe=pipe_logger:new("mapred-result"),
	ResultPipe=pipe_join:new([{WorkerId2,KeyFun},{WorkerId1,KeyFun}],ConsolePipe),
	%ResultPipe=pipe_logger:new("before-join", JoinPipe),
	Pg=#pg{pgs=[
				#pg{workers=[
							 #worker{
									 id=WorkerId1,
									 spawn_spec={fun()->
														 ?LOG_TRACE("PIPE_1",[]),
														 Pipe=new(WorkerId1,ResultPipe),
														 %Iter=iter:from_list([{"aa",1},{"b-k",2},{"c",3},{"ee",3},{"ffff",3}]),
														 Iter=iter_disk_log:new([{name,"BigData"},{mode, read_only}]),
														 {IterState,PipeState}=pipe:push_each(Pipe, Iter),
														 iter:close(IterState),
														 pipe:close(PipeState)
												 end}
									},
							 #worker{
									 id=WorkerId2,
									 spawn_spec={fun()->
														 ?LOG_TRACE("PIPE_2",[]),
														 Pipe=new(WorkerId2,ResultPipe),
														 %Iter=iter:from_list([{"aa",4},{"aa",2},{"c",35}]),
														 Iter=iter_disk_log:new([{name,"BigData"},{mode, read_only}]),
														 {IterState,PipeState}=pipe:push_each(Pipe, Iter),
														 iter:close(IterState),
														 pipe:close(PipeState)
												 end}
									}
							 ]}
			   ]}, 
	pgt:new(Pg).
